package com.dongdongshop.rocketscore;

import com.alibaba.dubbo.config.annotation.Reference;
import com.dongdongshop.pojo.TbUserScoreAccount;
import com.dongdongshop.service.UserScoreService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

@Component
@RocketMQMessageListener(topic = "order-topic",consumerGroup = "junjunjun")
public class ConsumerRocketScore implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {
    @Reference
    private UserScoreService userScoreService;
    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public void onMessage(String s) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);
                String[] split = new String(messageExt.getBody()).split(",");
                String totalScore = split[0];
                String str = totalScore.substring(0,totalScore.length()-3);
                int totalScoreInt = Integer.parseInt(str);
                String userId = split[1];
                try {

                    if (!redisTemplate.boundHashOps("order").hasKey(messageExt.getMsgId())) {

                        TbUserScoreAccount userScoreAccount = userScoreService.selectByUserid(userId);
                        TbUserScoreAccount tbUserScoreAccount = new TbUserScoreAccount();
                        tbUserScoreAccount.setUserId(userId);
                        tbUserScoreAccount.setCreateTime(new Date());
                        tbUserScoreAccount.setUpdateTime(new Date());
                        if (userScoreAccount == null) {
                            tbUserScoreAccount.setTotalScore(totalScoreInt);
                            tbUserScoreAccount.setCreateTime(new Date());
                            userScoreService.insertInfo(tbUserScoreAccount);
                        } else {
                           int totalScore1 = userScoreAccount.getTotalScore();
                            tbUserScoreAccount.setTotalScore(totalScore1 + totalScoreInt);
                            tbUserScoreAccount.setUpdateTime(new Date());
                            userScoreService.updateByUserid(tbUserScoreAccount);
                        }

                        redisTemplate.boundHashOps("order").put(messageExt.getMsgId(), "success");
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                    if (messageExt.getReconsumeTimes()>3){
                        consumeConcurrentlyContext.setDelayLevelWhenNextConsume(-1);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
